package com.cw.realtime.common
package util

import io.circe.Json
import org.apache.hadoop.hbase.{CellUtil, TableName}
import org.apache.hadoop.hbase.client._

import java.io.IOException
import java.nio.charset.Charset
import scala.collection.JavaConverters._

object HbaseUtil {

  @throws[IOException]
  def getConnection: Option[Connection] = Option(ConnectionFactory.createConnection)


  def closeConnection(conn: Option[Connection]): Unit =
    conn.foreach { c =>  if (!c.isClosed) c.close()}


  @throws[IllegalArgumentException]
  @throws[IOException]
  def createTable(conn: Connection, namespace: String, table: String, families: String*): Unit = {

    if (families.size < 1)
      throw new IllegalArgumentException("至少需要指定一个列簇")
    println(s"$namespace: $table")

    val admin = conn.getAdmin
    if (!admin.tableExists(TableName.valueOf(namespace, table))) {

      val tableDescBuild = TableDescriptorBuilder
        .newBuilder(TableName.valueOf(namespace, table))

      val columnFamilyDescriptors = families.map { name => ColumnFamilyDescriptorBuilder.newBuilder(name.getBytes).build() }.asJava
      tableDescBuild.setColumnFamilies(columnFamilyDescriptors)

      try
        admin.createTable(tableDescBuild.build())
      catch {
        case e: IOException => e.printStackTrace()
      }
    }
    admin.close()
  }

  @throws[IOException]
  def dropTable(conn: Connection, namespace: String, table: String): Unit = {
    val admin = conn.getAdmin

    try {
      val tb = TableName.valueOf(namespace, table)
      admin.disableTable(tb)
      admin.deleteTable(tb)
    } catch {
      case e: Exception => e.printStackTrace()
    }

    admin.close()
  }



  def getCells(conn: Connection, namespace: String, tableName: String, rowKey: String, family: String) = {
    val table = conn.getTable(TableName.valueOf(namespace, tableName))
    val get = new Get(rowKey.getBytes)
    val map = table.get(get).rawCells
      .map { cell => new String(CellUtil.cloneQualifier(cell)) -> new String(CellUtil.cloneValue(cell))}
      .toMap
    table.close()
    map
  }





  @throws[IOException]
  def putCells(
    conn: Connection, namespace: String, tableName: String, rowKey: String, family: String, data: Json
  ): Unit = {

    val table = conn.getTable(TableName.valueOf(namespace, tableName))
    val toPut = new Put(rowKey.getBytes)
    for {
      data <- data.asObject
      (column, v) <- data.toMap
      value = if (v.isString) v.asString.get else v.toString
    } {
      println(
        s"""********真的在插入数据了*******
          |${tableName} + ${family} + ${column} -> ${value}
          |
          |""".stripMargin)
      if (value != "null")
        toPut.addColumn(family.getBytes, column.getBytes, value.getBytes)
    }
    table.put(toPut)
  }


  @throws[IOException]
  def deleteCells(
    conn: Connection, namespace: String, tableName: String, rowKey: String
  ): Unit = {
    val table = conn.getTable(TableName.valueOf(namespace, tableName))
    val toDelete = new Delete(rowKey.getBytes)
    table.delete(toDelete)
  }


}
